#pragma once

#include <sys/timerfd.h>
#include <sys/eventfd.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <signal.h>
#include <cstdlib>
#include <cassert>
#include <cerrno>
#include <cstring>
#include <cstdint>
#include <vector>
#include <string>
#include <functional>
#include <unordered_map>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <memory>
#include <algorithm>
#include "log.hpp"
#include "Any.hpp"

const int MAX_EVENTS_SIZE = 1024;
const uint64_t DEFAULT_BUFFER_SIZE = 1024;
const int DEFAULT_BACKLOG = 1024;

class Buffer
{
public:
    Buffer()
        :_buffer(DEFAULT_BUFFER_SIZE), _reader_idx(0), _writer_idx(0)
    {}

    Buffer(const Buffer& other)
        :_buffer(DEFAULT_BUFFER_SIZE), _reader_idx(0), _writer_idx(0)
    {
        WriteAndPush(other);
    }
    
    //获取_buffer存储空间首地址
    const char* Begin() const {
        return &_buffer[0];
    }
    char* Begin() {
        return &_buffer[0];
    }
    //获取读位置地址
    char* GetReaderPos() {
        return Begin() + _reader_idx;
    }
    const char* GetReaderPos() const {
        return Begin() + _reader_idx;
    }
    //获取写位置地址
    char* GetWriterPos() {
        return Begin() + _writer_idx;
    }
    const char* GetWriterPos() const {
        return Begin() + _writer_idx;
    }

    //获取tail空闲空间
    const uint64_t GetTailIdleSpace() const {
        return _buffer.size() - _writer_idx;
    }
    //获取head空闲空间
    const uint64_t GetHeadIdleSpace() const {
        return _reader_idx;
    }
    //获取可读空间大小
    const uint64_t GetReadableSpace() const {
        return _writer_idx - _reader_idx;
    }
    //缓冲区为空？
    bool Empty() {
        return (GetReadableSpace() == 0);
    }
    
    //写入数据
    void Write(const void* data, const uint64_t len) //写一块内存空间
    {
        EnsureWriteableSpace(len);
        std::copy((char*)data, (char*)data + len, GetWriterPos());
    }
    void Write(const std::string& str) //写string
    {
        size_t len = str.size();
        EnsureWriteableSpace(len);
        std::copy(str.begin(), str.end(), GetWriterPos());
    }
    void Write(const Buffer& buf) //写Buffer
    {
        uint64_t len = buf.GetReadableSpace();
        EnsureWriteableSpace(len);
        const char* bufReaderPos = buf.GetReaderPos();
        std::copy(bufReaderPos, bufReaderPos + len, GetWriterPos());
    }

    //写入数据并刷新写指针
    void WriteAndPush(const void* data, const uint64_t len) {
        Write(data, len);
        WriterMove(len);
    }
    void WriteAndPush(const std::string& str) {
        Write(str);
        WriterMove(str.size());
    }
    void WriteAndPush(const Buffer& buf) {
        Write(buf);
        WriterMove(buf.GetReadableSpace());
    }

    //读取数据
    void Read(void* data, const uint64_t len) //读一块内存空间
    {
        assert(len <= GetReadableSpace());
        char* readerPos = GetReaderPos();
        std::copy(readerPos, readerPos + len, (char*)data);
    }
    std::string Read(const uint64_t len) //读一个字符串
    {
        assert(len <= GetReadableSpace());
        std::string str; str.resize(len);
        Read(&str[0], len);
        return str;
    }

    //读取数据并刷新读指针
    void ReadAndPop(void* data, const uint64_t len) {
        Read(data, len);
        ReaderMove(len);
    }
    std::string ReadAndPop(const uint64_t len) {
        std::string str = Read(len);
        ReaderMove(len);
        return str;
    }

    //读取一行数据: 读直到遇到换行符
    std::string GetLine() {
        char* CRLFPos = FindCRLF();
        std::string str;
        if (CRLFPos == nullptr) { return ""; }
        else {
            return Read(CRLFPos - GetReaderPos() + 1); // 把'\n'也读出来
        }
    }
  	//读取一行数据并刷新读指针
    std::string GetLineAndPop() {
        std::string str = GetLine();
        ReaderMove(str.size());
        return str;
    }

    //清空缓冲区
    void Clear() {
        _reader_idx = _writer_idx = 0;
    }

private:
    //读位置移动
    void ReaderMove(const uint64_t len) {
        assert(len <= GetReadableSpace());
        _reader_idx += len;
    }
    //写位置移动
    void WriterMove(const uint64_t len) {
        assert(len <= GetTailIdleSpace());
        _writer_idx += len;
    }

    //确保写入空间大小足够
    void EnsureWriteableSpace(const uint64_t len) {
        // 判断tail空间是否足够
        uint64_t TailSpace = GetTailIdleSpace();
        // 如果足够 则返回
        if (len <= TailSpace) { return; }
        
        // 如果不够 则判断可写空间大小是否足够
        uint64_t WriteableSpace = GetHeadIdleSpace() + TailSpace;
        // 如果足够 则移动数据
        if (len <= WriteableSpace) {
            uint64_t readableSpace = GetReadableSpace();
            std::copy(GetReaderPos(), GetReaderPos() + readableSpace, Begin());
            // 更新读写位置
            _reader_idx = 0;
            _writer_idx = readableSpace;
        }
        else { // 如果不够 则扩容
            _buffer.resize(_writer_idx + GetTailIdleSpace() + len);
        }
        return;
    }

    //寻找换行符
    char* FindCRLF() {
        return (char*)memchr(GetReaderPos(), '\n', GetReadableSpace());
    }

private:
    /* 不直接使用string：兼容包含'\0'的数据流 */
    std::vector<char> _buffer;
    uint64_t _reader_idx;
    uint64_t _writer_idx;
};

class Socket
{
public:
    Socket() : _sockfd(-1) {}

    Socket(int sockfd) : _sockfd(sockfd) {}
    // 创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ns_log::lg(ns_log::ERROR, "socket err");
            return false;
        }
        return true;
    }

    int Fd() { return _sockfd; }

    // 绑定地址信息
    bool Bind(const uint16_t port, const std::string &ip = "0.0.0.0")
    {
        struct sockaddr_in local;
        bzero(&local, sizeof(local));
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = inet_addr(ip.c_str());
        if (bind(_sockfd, (const sockaddr *)&local, sizeof(local)) < 0)
        {
            ns_log::lg(ns_log::ERROR, "bind err");
            return false;
        }
        return true;
    }
    // 开始监听
    bool Listen(int backlog = DEFAULT_BACKLOG)
    {
        if (listen(_sockfd, backlog) < 0)
        {
            ns_log::lg(ns_log::ERROR, "listen err");
            return false;
        }
        return true;
    }
    // 向服务器发起连接
    bool Connect(const uint16_t port, const std::string &ip)
    {
        struct sockaddr_in peer;
        bzero(&peer, sizeof(peer));
        peer.sin_family = AF_INET;
        peer.sin_port = htons(port);
        peer.sin_addr.s_addr = inet_addr(ip.c_str());
        if (connect(_sockfd, (const sockaddr *)&peer, sizeof(peer)) < 0)
        {
            ns_log::lg(ns_log::ERROR, "connect err");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept()
    {
        int newFd = accept(_sockfd, nullptr, nullptr);
        if (newFd < 0)
        {
            ns_log::lg(ns_log::ERROR, "accept err");
            return -1;
        }
        return newFd;
    }
    // 接收数据
    ssize_t Recv(void *data, size_t len, int flag = 0)
    {
        ssize_t ret = recv(_sockfd, data, len, flag);
        if (ret <= 0)
        {
            // 1.非阻塞情况下接收缓冲区为空
            // 2.阻塞等待时被信号打断
            if (errno == EAGAIN || errno == EINTR) {
                return 0;
            }
            else {
                ns_log::lg(ns_log::ERROR, "recv err");
                return -1;
            }
        }
        return ret;
    }
    // 发送数据
    ssize_t Send(const void *data, size_t len, int flag = 0)
    {
        if (len == 0) { return 0; }
        ssize_t ret = send(_sockfd, data, len, flag);
        if (ret <= 0)
        {
            if (errno == EAGAIN || errno == EINTR) {
                return 0;
            }
            else
            {
                ns_log::lg(ns_log::ERROR, "send err");
                return -1;
            }
        }
        return ret;
    }
    // 非阻塞接收
    ssize_t RecvNoBlock(void *data, size_t len)
    {
        return Recv(data, len, MSG_DONTWAIT);
    }
    // 非阻塞发送
    ssize_t SendNoBlock(const void *data, size_t len)
    {
        return Send(data, len, MSG_DONTWAIT);
    }
    // 关闭套接字
    void Close()
    {
        if (_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }
    // 创建服务器
    bool ServerCreate(const uint16_t port, const std::string &ip = "0.0.0.0", bool flag = true)
    {
        // 1.创建套接字 2.设置非阻塞 3.绑定 4.监听 5.地址&端口重用
        if (!Create()) { return false; }
        if (flag) {
            SetNonBlock();
        }
        if (!Bind(port, ip)) { return false; }
        if (!Listen()) { return false; }
        SetReuse();
        return true;
    }
    // 创建客户端连接
    bool ClinetCreate(const std::string &ip, const uint16_t port)
    {
        // 1.创建套接字 2.连接服务器
        if (!Create()) {
            return false;
        }
        if (!Connect(port, ip)) {
            return false;
        }
        return true;
    }
    // 设置套接字选项 -> 地址复用
    void SetReuse()
    {
        int optVal = 1;
        if(setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, \
                (void*)&optVal, sizeof(optVal)) < 0) {
            ns_log::lg(ns_log::ERROR, "setsockopt err");
        }
    }
    // 设置套接字选项 -> 非阻塞
    void SetNonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }

private:
    int _sockfd;
};

class Poller;
class EventLoop;

class Channel
{
private:
    using EventCallBack = std::function<void()>;

public:
    Channel(int fd, EventLoop *loop)
        : _fd(fd), _loop(loop), _events(0), _revents(0)
    { }

    const int Fd() const {
        return _fd;
    }
    // 获取监听事件
    const uint32_t Events() const {
        return _events;
    }
    // 设置已经就绪的事件
    void ReventSet(const uint32_t events) {
        _revents = events;
    }
    // 是否可读？
    bool Readable() const { return (_events & EPOLLIN); }
    // 是否可写？
    bool Writeable() const { return (_events & EPOLLOUT); }
    // 添加可读事件监控
    void ReadEnable()
    {
        _events |= EPOLLIN;
        Update(); // 更新/添加对当前对象的监控
    }
    // 添加可写事件监控
    void WriteEnable()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 取消可读事件监控
    void ReadDisable()
    {
        _events &= ~(EPOLLIN);
        Update();
    }
    // 取消可写事件监控
    void WriteDisable()
    {
        _events &= ~(EPOLLOUT);
        Update();
    }
    // 更新/添加针对当前对象的监控
    void Update();
    // 移除针对当前对象的监控
    void Remove();
    // 取消所有事件监控
    void DisableAll() { _events = 0; Update(); }
    // 设置回调
    void ReadCbSet(const EventCallBack &cb) { _read_cb = cb; }
    void WriteCbSet(const EventCallBack &cb) { _write_cb = cb; }
    void CloseCbSet(const EventCallBack &cb) { _close_cb = cb; }
    void ErrorCbSet(const EventCallBack &cb) { _error_cb = cb; }
    void EventsCbSet(const EventCallBack &cb) { _events_cb = cb; }
    // 执行事件回调
    void Handler()
    {
        // 1.读事件就绪 2.对方关闭连接 -> 读取缓冲区剩余内容 3.收到带外数据
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI)) {
            if (_events_cb) { _events_cb(); }
            if (_read_cb) { _read_cb(); }
        }
        // 写入的时候可能关闭连接……
        if (_revents & EPOLLOUT) {
            if (_events_cb) { _events_cb(); }
            if (_write_cb) { _write_cb(); }
        } // 一次只进行一次有可能关闭连接的操作
        else if (_revents & EPOLLHUP) {
            if (_events_cb) { _events_cb(); }
            if (_close_cb) { _close_cb(); }
        }
        else if (_revents & EPOLLERR) {
            if (_events_cb) { _events_cb(); }
            if (_error_cb) { _error_cb(); }
        } else {}
    }

private:
    int _fd;
    EventLoop *_loop; // 所绑定的EventLoop对象
    uint32_t _events; //所关心的事件
    uint32_t _revents; //已经就绪的事件
    EventCallBack _read_cb;
    EventCallBack _write_cb;
    EventCallBack _close_cb;
    EventCallBack _error_cb;
    EventCallBack _events_cb;
};

const int EPOLL_SIZE = 128;

class Poller
{
public:
    // 创建epoll模型
    Poller()
    {
        _epfd = epoll_create(EPOLL_SIZE);
        if (_epfd < 0) {
            ns_log::lg(ns_log::FATAL, "epoll create err");
            abort();
        }
    }
    // 添加/更新描述符事件
    void Update(Channel *ch)
    {
        if (HasChannel(ch)) { // 存在 修改
            EpollUpdate(ch, EPOLL_CTL_MOD);
        }
        else { // 不存在 添加
            EpollUpdate(ch, EPOLL_CTL_ADD);
            _events.insert(std::make_pair(ch->Fd(), ch));
        }
    }
    // 移除描述符事件
    void Remove(Channel *ch)
    {
        auto iter = _events.find(ch->Fd());
        if (iter != _events.end())
        {
            //在epoll模型中移除
            EpollUpdate(ch, EPOLL_CTL_DEL);
            //在_events中删除
            _events.erase(iter);
        }
    }
    // 开始监控
    void Poll(std::vector<Channel *> *actives)
    {
        //int epoll_wait(int epfd, struct epoll_event *events, \ 
        //      int maxevents, int timeout);
        // 阻塞监控
        int n = epoll_wait(_epfd, _evs, MAX_EVENTS_SIZE, -1);
        if (n < 0)
        {
            ns_log::lg(ns_log::ERROR, "epoll_wait err");
            abort();
        }
        for (int i = 0; i < n; ++i)
        {
            auto iter = _events.find(_evs[i].data.fd);
            assert(iter != _events.end());
            // 设置就绪事件
            iter->second->ReventSet(_evs[i].events);
            // 添加结果
            actives->push_back(iter->second);
        }
    }

private:
    // epoll操作
    void EpollUpdate(const Channel *ch, int op)
    {
        // int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
        int fd = ch->Fd();
        struct epoll_event ev;
        bzero(&ev, sizeof(ev));
        ev.data.fd = fd;
        ev.events = ch->Events();
        if (epoll_ctl(_epfd, op, fd, &ev) < 0) {
            ns_log::lg(ns_log::ERROR, "epoll_ctl err");
        }
        return;
    }
    // 描述符事件是否存在？
    bool HasChannel(const Channel *ch)
    {
        return (_events.find(ch->Fd()) != _events.end());
    }

private:
    int _epfd;
    struct epoll_event _evs[MAX_EVENTS_SIZE];
    std::unordered_map<int, Channel *> _events; // fd - Channel
};

class TimerTask
{
private:
    using taskFunc = std::function<void()>;
    using realeaseFunc = std::function<void()>;

public:
    TimerTask(uint64_t id, uint32_t delayTime, const taskFunc& cb)
        :_id(id), _timeout(delayTime), _task_cb(cb), _canceled(false)
    {}
    //设置定时任务执行回调
    void SetRealeaseFunc(const realeaseFunc& realease_cb)
    {
        _realease_cb = realease_cb;
    }

    uint32_t DelayTime() {
        return _timeout;
    }

    void Cancel() {
        _canceled = true;
    }

    //在析构中执行定时任务和_realease_cb
    ~TimerTask()
    {
        if (!_canceled) { _task_cb(); }
        _realease_cb();
    }

private:
    uint64_t _id; //定时任务对象id
    uint32_t _timeout; //定时任务超时时间
    taskFunc _task_cb; //定时任务回调
    realeaseFunc _realease_cb; //定时任务执行回调
    bool _canceled; //任务是否被取消
};

class TimerWheel
{
private:
    using taskFunc = std::function<void()>;
    using realeaseFunc = std::function<void()>;

public:
    TimerWheel(EventLoop* loop)
        :_tick(0), _capacity(60), _wheel(_capacity),
        _loop(loop), _timerfd(TimerfdCreate()),
        _timerfd_ch(new Channel(_timerfd, _loop))
    {
        //设置_timerfd_ch的读事件回调并启动读事件监控
        _timerfd_ch->ReadCbSet(std::bind(&TimerWheel::TimeOut, this));
        _timerfd_ch->ReadEnable();
    }
    //避免多线程操作TimerWheel时的并发问题 -> 将操作放到同一个线程中执行
    void TimerTaskAdd(uint64_t id, uint32_t delayTime, const taskFunc& cb);
    void TimerTaskRefresh(uint64_t fd);
    void TaskCancel(uint64_t id);

    //有线程安全问题，只限在模块内使用
    bool HasTimerTask(uint64_t id)
    {
        return (_timerTasks.find(id) != _timerTasks.end());
    }

    ~TimerWheel() {}

private:
    using taskPtr = std::shared_ptr<TimerTask>;
    using taskWeakPtr = std::weak_ptr<TimerTask>;

private:
    void TaskRealeaseFunc(uint64_t id)
    {
        //在定时任务管理表中移除对应的任务
        auto iter = _timerTasks.find(id);
        if (iter == _timerTasks.end()) { return; }
        _timerTasks.erase(iter);
    }
    //添加定时任务
    void TimerTaskAddInLoop(uint64_t id, uint32_t delayTime, const taskFunc& cb)
    {
        //创建定时任务管理对象
        taskPtr taskPt(new TimerTask(id, delayTime, cb));
        //设置任务执行回调
        taskPt->SetRealeaseFunc(std::bind(&TimerWheel::TaskRealeaseFunc, this, id));
        //将定时任务进行管理
        taskWeakPtr taskWpt(taskPt);
        // _timerTasks.insert(std::make_pair(id, taskWpt)); //Cancel之后立即Add相同id的任务，任务不会被Add成功
        _timerTasks[id] = taskWpt; //Cancel之后立即Add相同id的任务，任务可以被Add(替换)成功
        //将定时任务管理对象添加到时间轮
        int pos = (_tick + delayTime) % _capacity;
        _wheel[pos].push_back(taskPt);
    }
    //延时定时任务
    void TimerTaskRefreshInLoop(uint64_t id)
    {
        auto iter = _timerTasks.find(id);
        if (iter == _timerTasks.end()) { return; }
        taskPtr taskPt = iter->second.lock(); //获取weak_ptr对应的shared_ptr对象
        //计算新的时间轮位置，并将新的任务管理对象加入时间轮
        int pos = (_tick + taskPt->DelayTime()) % _capacity;
        _wheel[pos].push_back(taskPt);
    }
    //定时任务取消
    //被取消的定时任务不会被立即回调和realease,而是在超时时进行
    void TaskCancelInLoop(uint64_t id)
    {
        auto iter = _timerTasks.find(id);
        if (iter == _timerTasks.end()) { return; }
        taskPtr taskpt = iter->second.lock();
        //如果任务管理对象未过期，就取消对应的任务
        if (taskpt) { taskpt->Cancel(); }
    }

    int TimerfdRead()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, 8);
        if(ret < 0) {
            ns_log::lg(ns_log::FATAL, "timerfd read err");
            abort();
        }
        return times;
    }
    //启动时间轮
    //需要每秒钟执行一次
    void TickTack()
    {
        _tick = (_tick + 1) % _capacity;
        _wheel[_tick].clear(); // run timer task
    }

    void TimeOut()
    {
        int times = TimerfdRead(); //读取timerfd
        while (times--) {
            TickTack(); //run timer task
        }
    }

    static int TimerfdCreate()
    {
        //创建timerfd并设定通知周期
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0) {
            ns_log::lg(ns_log::FATAL, "timerfd_create err");
            abort();
        }
        struct itimerspec timerspec;
        bzero(&timerspec, sizeof(timerspec));
        timerspec.it_value.tv_sec = 1; //设置第一次超时时间
        timerspec.it_interval.tv_sec = 1; //设置后续超时时间
        if(timerfd_settime(timerfd, 0, &timerspec, nullptr) < 0) {
            ns_log::lg(ns_log::FATAL, "timerfd_settime err");
            abort();
        }
        return timerfd;    
    }

private:
    uint32_t _tick; //时间轮指针
    uint32_t _capacity; //表盘大小(最大超时时间)
    std::vector<std::vector<taskPtr>> _wheel; //时间轮
    //定时任务管理
    std::unordered_map<uint64_t, taskWeakPtr> _timerTasks; //任务id - timerTask

    EventLoop* _loop; // 绑定的EventLoop对象
    int _timerfd;
    std::unique_ptr<Channel> _timerfd_ch; // 对_timerfd进行管理
};

class EventLoop
{
private:
    using Functor = std::function<void()>;
    /*-------------------------------
        from class TimerWheel
    -------------------------------*/
    using taskFunc = std::function<void()>;
    using realeaseFunc = std::function<void()>;

public:
    EventLoop()
        :_thread_id(std::this_thread::get_id()),
        _timer_wheel(this),
        _event_fd(EventFdCreate()),
        _event_channel(new Channel(_event_fd, this)) //to do
    {
        //为_event_channel设置读事件回调并启动读事件
        _event_channel->ReadCbSet(std::bind(&EventLoop::EventFdReadCb, this));
        _event_channel->ReadEnable();
    }
    //判断当前执行流是否在EventLoop所在的线程中
    bool IsInLoop() {
        return (_thread_id == std::this_thread::get_id());
    }
    void InLoopAssert() {
        assert(_thread_id == std::this_thread::get_id());
    }
    //使任务在EventLoop所在的线程中执行
    void RunInLoop(const Functor& cb)
    {
        if (IsInLoop()) { cb(); }
        else {  TaskPush(cb); }
    }
    //将任务放入任务池中
    void TaskPush(const Functor& cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(cb);
        }
        EventFdWakeUp(); //通知poller，唤醒其可能处于的阻塞状态
    }
    //增加/修改对Channel的监控
    void EventUpdate(Channel* ch) {
        _poller.Update(ch);
    }
    //移除对Channel的监控
    void EventRemove(Channel* ch) {
        _poller.Remove(ch);
    }
    //TimerTask操作
    void TimerTaskAdd(uint64_t id, uint32_t delayTime, const taskFunc& cb) {
        return _timer_wheel.TimerTaskAdd(id, delayTime, cb);
    }
    void TimerTaskRefresh(uint64_t id) {
        return _timer_wheel.TimerTaskRefresh(id);
    }
    void TimerTaskCancel(uint64_t id) {
        return _timer_wheel.TaskCancel(id);
    }
    //任务是否在时间轮中
    bool HasTimerTask(uint64_t id) {
        return _timer_wheel.HasTimerTask(id);
    }

    //启动组件
    void Start() {
        while (true)
        {
            //1.事件监控
            std::vector<Channel*> actives;
            _poller.Poll(&actives); //可能阻塞到此处，导致RunAllTask()无法执行
            //2.事件处理
            for (const auto& active : actives) {
                active->Handler();
            }
            //3.执行(任务池中的)任务
            RunAllTask();
        }
    }

private:
    //创建event fd
    static int EventFdCreate()
    {
        int event_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if(event_fd < 0) {
            ns_log::lg(ns_log::FATAL, "EventFdCreate err");
            abort();
        }
        return event_fd;
    }
    //EventFd读事件回调
    void EventFdReadCb()
    {
        uint64_t cnt = 0;
        ssize_t n = read(_event_fd, &cnt, 8);
        if (n < 0) {
            if (errno == EINTR || errno == EAGAIN) { return; }
            else {
                ns_log::lg(ns_log::FATAL, "EventFdReadCb err");
                abort();
            }
        }
    }
    //唤醒poller可能出现的阻塞状态: 向EventFd中写入
    void EventFdWakeUp()
    {
        uint64_t cnt = 0;
        ssize_t n = write(_event_fd, &cnt, 8);
        if (n < 0) {
            if (errno == EINTR || errno == EAGAIN) { return; }
            else {
                ns_log::lg(ns_log::FATAL, "EventFdWakeUp err");
                abort();
            }
        }
    }
    //执行任务池中的所有任务
    void RunAllTask()
    {
        std::vector<Functor> functors;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            functors.swap(_tasks);
        }
        for (auto functor : functors) {
            functor();
        }
    }

private:
    std::thread::id _thread_id; //EventLoop所绑定的线程id
    Poller _poller; //事件监控
    TimerWheel _timer_wheel;
    int _event_fd; //取消epoll的阻塞
    std::unique_ptr<Channel> _event_channel; //管理eventfd
    std::vector<Functor> _tasks; //任务池
    std::mutex _mutex; //mutex for _tasks
};

class LoopThread
{
public:
    LoopThread()
        :_loop(nullptr), _thread(&LoopThread::ThreadEntry, this)
    { }

    EventLoop* GetLoop()
    {
        // GetLoop时_loop可能未实例化完成
        EventLoop* loop;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _cond_var.wait(_lock, [&](){ return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }

private:
    void ThreadEntry()
    {
        // 实例化EventLoop对象，启动_loop
        EventLoop loop; // 创建栈中的临时对象：希望loop的生命周期随EventLoop的工作周期
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _loop = &loop;
            _cond_var.notify_all(); // 唤醒可能的线程阻塞
        }
        _loop->Start(); // 启动_loop
    }

private:
    std::mutex _mutex;
    std::condition_variable _cond_var;
    EventLoop* _loop;
    std::thread _thread;
};

class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop* base_loop)
        :_thread_count(0), _next_loop_idx(0), _base_loop(base_loop)
    {}

    void SetThreadCount(int cnt) {
        _thread_count = cnt;
    }
    // 创建线程
    void Fill()
    {
        if (_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for (int i = 0; i < _thread_count; ++i)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }
    // 返回EventLoop
    EventLoop* NextLoop()
    {
        // 当_thread_count为0时，相当于单线程单Reactor模型
        if (_thread_count == 0) {
            return _base_loop;
        }
        // RR轮转
        _next_loop_idx = (_next_loop_idx + 1) % _thread_count;
        return _loops[_next_loop_idx];
    }

private:
    int _thread_count;
    int _next_loop_idx;
    EventLoop* _base_loop; // listen loop
    std::vector<LoopThread*> _threads;
    std::vector<EventLoop*> _loops;
};

enum ConnectStates
{
    CONNECTING, //连接已建立，尚未进行establish配置
    CONNECTED, //连接配置完毕
    DISCONNECTING, //准备关闭连接
    DISCONNECTED //连接已关闭
};

class Connection;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    using ConnectedCb = std::function<void(const PtrConnection&)>;
    using MessageCb = std::function<void(const PtrConnection&, Buffer*)>;
    using ClosedCb = std::function<void(const PtrConnection&)>;
    using EventCb = std::function<void(const PtrConnection&)>;

public:
    Connection(EventLoop* loop, uint64_t conn_id, int sock_fd)
        :_conn_id(conn_id), _sock_fd(sock_fd), _loop(loop), _sock(sock_fd), \
        _channel(_sock_fd, _loop), _enable_inactive_release(false), _state(CONNECTING)
    {
        _channel.ReadCbSet(std::bind(&Connection::HandleRead, this));
        _channel.WriteCbSet(std::bind(&Connection::HandleWrite, this));
        _channel.CloseCbSet(std::bind(&Connection::HandlerClose, this));
        _channel.ErrorCbSet(std::bind(&Connection::HandlerErr, this));
        _channel.EventsCbSet(std::bind(&Connection::HandlerEvent, this));
    }
    //是否处于Connected状态
    bool IsConnected() {
        return (_state == CONNECTED);
    }
    // 配置连接 -> 启动Channel的读监控，执行用户回调
    void Establish() {
        _loop->RunInLoop(std::bind(&Connection::EstablishInLoop, this));
    }
    // 设置Connection的回调
    void SetConnectedCb(const ConnectedCb& connected_cb) {
        _connected_cb = connected_cb;
    }
    void SetMessageCb(const MessageCb& msg_cb) {
        _msg_cb = msg_cb;
    }
    void SetClosedCb(const ClosedCb& cls_cb) {
        _cls_cb = cls_cb;
    }
    void SetSvrCloseCb(ClosedCb svr_cls_cb) {
        _svr_cls_cb = svr_cls_cb;
    }
    void SetEventCb(const EventCb& event_cb) {
        _event_cb = event_cb;
    }
    // 设置上下文
    void SetContext(const Any& context) {
        _context = context;
    }
    // 获取上下文
    Any* GetContext() {
        return &_context;
    }
    // 获取fd
    int Fd() {
        return _sock_fd;
    }
    // 获取Connection id
    uint64_t Id() {
        return _conn_id;
    }
    // 发送数据
    // 上层传入的data可能是一块临时空间，且Send有可能被被放入任务池而无法被立即执行，
    // 导致实际写入时data的空间已被释放 -> 解决：额外构造buffer对象
    void Send(const char* data, size_t len) {
        Buffer buff;
        buff.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buff)));
    }
    // 给用户提供的连接关闭接口
    void ShutDown() {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    // 启动非活跃连接关闭功能
    void InactiveRealeaseEnable(const int sec) {
        _loop->RunInLoop(std::bind(&Connection::InactiveReleaseEnableInLoop, this, sec));
    }
    // 关闭非活跃连接关闭功能
    void InactiveRealeaseDisable() {
        _loop->RunInLoop(std::bind(&Connection::InactiveReleaseDisableInLoop, this));
    }
    // 协议切换 -> 切换回调函数和处理上下文
    // 需要被立即执行: 防止新事件触发后使用旧协议 -> 必须在EventLoop所在的线程中执行
    void Toggle(const Any& context, const ConnectedCb& connected_cb, const MessageCb& msg_cb, \
                    const ClosedCb& cls_cb, const EventCb& event_cb)
    {
        _loop->InLoopAssert();
        _loop->RunInLoop(std::bind(&Connection::ToggleInLoop, this, context, connected_cb, \
                                        msg_cb, cls_cb, event_cb));
    }
                    
    ~Connection() {
        // for debug
        std::cout << "connection release: " << this << std::endl;
    }

private:
    /*--------------------------
        Channel的五个事件回调
    --------------------------*/
    // 读事件回调：将数据放入输入缓冲区，调用业务处理回调_msg_cb
    void HandleRead()
    {
        // 1.从Socket接收数据 并放入输入缓冲区
        char buff[65536] = { 0 };
        //非阻塞接收
        ssize_t n = _sock.RecvNoBlock(buff, 65535);
        if (n < 0) {
            ns_log::lg(ns_log::ERROR, "recv from sock err");
            return ShutDown();
        }
        _in_buff.WriteAndPush(buff, n);
        // 2.调用业务处理回调
        if (_in_buff.GetReadableSpace() > 0) {
            //shared_from_this(): 获取基于当前对象的shared_ptr对象
            return _msg_cb(shared_from_this(), &_in_buff);
        }
    }
    // 写事件回调：将发送缓冲区的数据进行发送
    void HandleWrite()
    {
        uint64_t readSz = _out_buff.GetReadableSpace();
        char* data = new char[readSz];
        //从_out_buff读取数据
        _out_buff.ReadAndPop(data, readSz);
        //通过_socket发送数据
        ssize_t n = _sock.SendNoBlock(data, readSz);
        delete[] data;
        // 写入出错
        if (n < 0) {
            //输入缓冲区有残余数据 -> 进行处理并释放连接
            if (_in_buff.GetReadableSpace() > 0) {
                _msg_cb(shared_from_this(), &_in_buff);
            }
            return Release();
        }
        //缓冲区中没有可读数据
        if (_out_buff.GetReadableSpace() == 0)
        {
            _channel.WriteDisable(); //关闭读事件监控
            //如果连接处于连接待关闭状态且缓冲区没有数据，则释放连接
            if (_state == DISCONNECTING) {
                return Release();
            }
        }
    }

    void HandlerClose()
    {
        // 如果有残余数据 -> 进行处理
        if (_in_buff.GetReadableSpace() > 0) {
            _msg_cb(shared_from_this(), &_in_buff);
        }
        return Release(); // 释放连接
    }

    void HandlerErr() {
        return HandlerClose();
    }

    void HandlerEvent()
    {
        // 1.刷新连接活跃度
        if (_enable_inactive_release) {
            _loop->TimerTaskRefresh(_conn_id);
        }
        // 2.执行用户的回调
        if (_event_cb) {
            _event_cb(shared_from_this());
        }
    }
    void EstablishInLoop()
    {
        // 1.修改连接状态
        assert(_state == CONNECTING);
        _state = CONNECTED;
        // 2.启动可读事件
        _channel.ReadEnable();
        // 3.调用用户回调
        if (_connected_cb) { _connected_cb(shared_from_this()); }
    }
    // 连接释放
    void ReleaseInLoop()
    {
        // 1.修改连接状态
        _state = DISCONNECTED;
        // 2.移除事件监控
        _channel.Remove();
        // 3.关闭socket描述符
        _sock.Close();
        // 4.取消定时任务
        if (_loop->HasTimerTask(_conn_id)) {
            InactiveReleaseDisableInLoop();
        }
        // 5.执行用户回调
        if (_cls_cb) { _cls_cb(shared_from_this()); }
        // 6.移除服务器内部的管理信息
        if (_svr_cls_cb) { _svr_cls_cb(shared_from_this()); }
    }

    void SendInLoop(Buffer& buf)
    {
        // 如果连接处于DISCONNECTED状态则do nothing
        if (_state == DISCONNECTED) { return; }
        // 将数据放入输出缓冲区，并启动写事件监控
        _out_buff.WriteAndPush(buf);
        if (!_channel.Writeable()) {
            _channel.WriteEnable();
        }
    }
    // 不直接释放连接，而是放到任务池中执行
    void Release()
    {
        _loop->RunInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }

    void ShutDownInLoop()
    {
        // 修改连接状态
        _state = DISCONNECTING;
        // 输入缓冲区有数据 -> 进行处理
        if (_in_buff.GetReadableSpace() > 0) {
            if (_msg_cb) { _msg_cb(shared_from_this(), &_in_buff); }
        }
        // 输出缓冲区有数据 -> 启动可读事件监控
        if (_out_buff.GetReadableSpace() > 0) {
            if (!_channel.Writeable()) {
                _channel.WriteEnable();
            }
        }
        // 输出缓冲区为空 -> 释放连接
        if (_out_buff.GetReadableSpace() == 0) {
           Release(); 
        }
    }

    void InactiveReleaseEnableInLoop(int sec)
    {
        // 修改判断标志
        _enable_inactive_release = true;
        // 添加/刷新定时任务
        if (_loop->HasTimerTask(_conn_id)) {
            _loop->TimerTaskRefresh(_conn_id);
        }
        else {
            _loop->TimerTaskAdd(_conn_id, sec, std::bind(&Connection::Release, this));
        }
    }

    void InactiveReleaseDisableInLoop()
    {
        // 修改判断标志
        _enable_inactive_release = false;
        // 取消定时任务
        if (_loop->HasTimerTask(_conn_id)) {
            _loop->TimerTaskCancel(_conn_id);
        }
    }

    void ToggleInLoop(const Any& context, const ConnectedCb& connected_cb, \
            const MessageCb& msg_cb, const ClosedCb& cls_cb, const EventCb& event_cb)
    {
        SetContext(context);
        _connected_cb = connected_cb;
        _msg_cb = msg_cb;
        _cls_cb = cls_cb;
        _event_cb = event_cb;
    }

private:
    uint64_t _conn_id; //Connection id 同时作为 timer id
    int _sock_fd; // 连接关联的fd
    EventLoop* _loop; // 关联的EventLoop
    Socket _sock; // 套接字管理
    Channel _channel; // 连接和事件管理
    Buffer _in_buff; // 输入缓冲区
    Buffer _out_buff; // 输出缓冲区
    Any _context; // 请求的接收处理上下文
    bool _enable_inactive_release; // 是否启动非活跃连接的释放功能
    ConnectStates _state; // 连接状态
    
    ConnectedCb _connected_cb; //连接建立回调
    MessageCb _msg_cb; //信息处理回调
    ClosedCb _cls_cb; //连接关闭回调
    EventCb _event_cb; //任意事件回调

    ClosedCb _svr_cls_cb; //组件内部的连接关闭回调
};

class Acceptor
{
private:
    using AcceptCb = std::function<void(int)>;

public:
    Acceptor(EventLoop* loop, int port)
        :_loop(loop), _listen_socket(ServerCreate(port)),
        _listen_channel(_listen_socket.Fd(), _loop)
    {
        _listen_channel.ReadCbSet(std::bind(&Acceptor::ReadHandler, this));
    }

    void Listen() {
        _listen_channel.ReadEnable();
    }

    void AcceptCbSet(const AcceptCb& cb) {
        _accept_cb = cb;
    }

private:
    int ServerCreate(const int port)
    {
        if(!_listen_socket.ServerCreate(port)) {
            ns_log::lg(ns_log::FATAL, "ServerCreate err");
            abort();
        }
        return _listen_socket.Fd();
    }

    void ReadHandler()
    {
        // 获取新连接
        int newfd = _listen_socket.Accept();
        if (newfd < 0) {
            ns_log::lg(ns_log::ERROR, "_listen_socket accept err");
            return;
        }
        // 调用回调
        if (_accept_cb) { _accept_cb(newfd); }
    }

private:
    Socket _listen_socket; // 监听套接字
    EventLoop* _loop; // 对监听套接字进行事件监控
    Channel _listen_channel; // 对监听套接字进行事件管理
    AcceptCb _accept_cb; // 对新连接进行处理的回调
};

class TcpServer
{
private:
    using Functor = std::function<void()>;

    using ConnectedCb = std::function<void(const PtrConnection&)>;
    using MessageCb = std::function<void(const PtrConnection&, Buffer*)>;
    using ClosedCb = std::function<void(const PtrConnection&)>;
    using EventCb = std::function<void(const PtrConnection&)>;

public:
    TcpServer(int port)
        :_port(port),
        _next_id(0),
        _acceptor(&_base_loop, _port),
        _enable_inactive_release(false),
        _pool(&_base_loop)
    {
        _acceptor.AcceptCbSet(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen();
    }

    void SetThreadCount(int cnt)
    {
        _pool.SetThreadCount(cnt);
        _pool.Fill();
    }

    void InactiveReleaseEnable(int timeout) {
        return _base_loop.RunInLoop(std::bind(&TcpServer::InactiveReleaseEnableInLoop, this, timeout));
    }
    void InactiveReleaseDisable() {
        return _base_loop.RunInLoop(std::bind(&TcpServer::InactiveReleaseDisableInLoop, this));
    }

    // 设置定时任务
    void RunAfter(int timeout, const Functor& func) {
        return _base_loop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, timeout, func));
    }

    void SetConnectedCb(const ConnectedCb& connected_cb) {
        _connected_cb = connected_cb;
    }
    void SetMessageCb(const MessageCb& msg_cb) {
        _msg_cb = msg_cb;
    }
    void SetClosedCb(const ClosedCb& cls_cb) {
        _cls_cb = cls_cb;
    }
    void SetEventCb(const EventCb& event_cb) {
        _event_cb = event_cb;
    }

    void Start() {
        return _base_loop.Start();
    }

private:
    // 封装Connection
    void NewConnection(int fd) {
        _next_id++;
        std::shared_ptr<Connection> conn(new Connection(_pool.NextLoop(), _next_id, fd));
        conn->SetConnectedCb(_connected_cb);
        conn->SetClosedCb(_cls_cb);
        conn->SetMessageCb(_msg_cb);
        conn->SetSvrCloseCb(std::bind(&TcpServer::ConnectionRemove, this, std::placeholders::_1));
        if (_enable_inactive_release) {
            conn->InactiveRealeaseEnable(_timeout);
        }
        conn->Establish();
        _conns.insert(std::make_pair(conn->Id(), conn));
    }
    
    void ConnectionRemoveInLoop(const PtrConnection& conn)
    {
        auto iter = _conns.find(conn->Id());
        if (iter != _conns.end()) {
            _conns.erase(iter);
        }
    }
    // 连接释放
    void ConnectionRemove(const PtrConnection& conn) {
        return _base_loop.RunInLoop(std::bind(&TcpServer::ConnectionRemoveInLoop, this, conn));
    }
    void InactiveReleaseEnableInLoop(int timeout)
    {
        _enable_inactive_release = true;
        _timeout = timeout;
    }
    void InactiveReleaseDisableInLoop() {
        _enable_inactive_release = false;
    }
    void RunAfterInLoop(uint32_t timeout, const Functor& func)
    {
        _next_id++;
        _base_loop.TimerTaskAdd(_next_id, timeout, func);
    }

private:
    int _port;
    uint64_t _next_id;
    EventLoop _base_loop; // _acceptor和_pool所关联的base EventLoop
    Acceptor _acceptor;
    bool _enable_inactive_release;
    int _timeout;
    LoopThreadPool _pool;
    std::unordered_map<uint64_t, PtrConnection> _conns; // 对Connection进行统一管理

    ConnectedCb _connected_cb;
    MessageCb _msg_cb;
    ClosedCb _cls_cb;
    EventCb _event_cb;
};

class NetWork
{
public:
    NetWork() { signal(SIGPIPE, SIG_IGN); }
};

static NetWork nw;

/*--------------------------
    from class `Channel`
--------------------------*/
// 更新/添加针对当前对象的监控
void Channel::Update() { _loop->EventUpdate(this); }
// 移除针对当前对象的监控
void Channel::Remove() { _loop->EventRemove(this); }

/*--------------------------
    from class `TimerWheel`
--------------------------*/
//避免多线程操作TimerWheel时的并发问题 -> 将操作放到同一个线程中执行
void TimerWheel::TimerTaskAdd(uint64_t id, uint32_t delayTime, const taskFunc& cb) {
    _loop->RunInLoop(std::bind(&TimerWheel::TimerTaskAddInLoop, this, id, delayTime, cb));
}

void TimerWheel::TimerTaskRefresh(uint64_t fd) {
    _loop->RunInLoop(std::bind(&TimerWheel::TimerTaskRefreshInLoop, this, fd));
}

void TimerWheel::TaskCancel(uint64_t id) {
    _loop->RunInLoop(std::bind(&TimerWheel::TaskCancelInLoop, this, id));
}
